package test.zookeeper;

import java.io.IOException;
import java.util.Collections;
import java.util.List;
import java.util.concurrent.CountDownLatch;

import org.apache.commons.collections4.CollectionUtils;
import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.KeeperException;
import org.apache.zookeeper.WatchedEvent;
import org.apache.zookeeper.Watcher;
import org.apache.zookeeper.ZooKeeper;
import org.apache.zookeeper.Watcher.Event.EventType;
import org.apache.zookeeper.Watcher.Event.KeeperState;
import org.apache.zookeeper.ZooDefs.Ids;
import org.apache.zookeeper.data.Stat;

public class Zqueue {
	
	private ZooKeeper zookeeper;
	
	public Zqueue() {
		try {
			zookeeper = new ZooKeeper("127.0.0.1:2181", 3000, new Watcher() {
				@Override
				public void process(WatchedEvent event) {
					String path = event.getPath();
					EventType type = event.getType();
					KeeperState state = event.getState();
					System.out.println("path=" + path + ", type=" + type + ", state=" + state);
				}
			});
			Stat stat = zookeeper.exists("/queue", null);
			if(stat==null) {
				String result = zookeeper.create("/queue", null, Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
				System.out.println(result);
			}
		} catch (IOException e) {
			e.printStackTrace();
		} catch (KeeperException e) {
			e.printStackTrace();
		} catch (InterruptedException e) {
			e.printStackTrace();
		}
	}
	
	public void offer(String queueName, String str) {
		String key = "/queue/" + queueName;
		try {
			Stat stat = zookeeper.exists(key, null);
			if(stat==null) {
				zookeeper.create(key, null, Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
			}
			String result = zookeeper.create(key + "/index_", str.getBytes(), Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT_SEQUENTIAL);
			System.out.println(result + "," + str);
		} catch (KeeperException e) {
			e.printStackTrace();
		} catch (InterruptedException e) {
			e.printStackTrace();
		}
	}
	
	public String take(String queueName) throws KeeperException, InterruptedException {
		String key = "/queue/" + queueName;
		while(true) {
			CountDownLatch cdl = new CountDownLatch(1);
			List<String> list = null;
			try {
				list = zookeeper.getChildren(key, new Watcher() {
					@Override
					public void process(WatchedEvent event) {
						if(event.getType()==EventType.NodeChildrenChanged) {
							cdl.countDown();
						}
					}
				});
				System.out.println("list=" + list);
			} catch (KeeperException.NoNodeException e) {
				zookeeper.create(key, null, Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
			}
			
			if(CollectionUtils.isEmpty(list)) {
				cdl.await();
				continue;
			}
			Collections.sort(list);
			for(int i=0;i<list.size();i++) {
				String path = key + "/" + list.get(i);
				try {
					byte[] bytes = zookeeper.getData(path, false, null);
					zookeeper.delete(path, -1);
					return new String(bytes);
				} catch (KeeperException.NoNodeException e) {
					//有其它客户端先把该节点删了，已经出队
				}
				
			}
		}
	}
	
	public static void main(String[] args) {
		new Thread() {
			@Override
			public void run() {
				Zqueue queue = new Zqueue();
				try {
					String str = queue.take("hello");
					System.out.println(str);
				} catch (KeeperException e) {
					e.printStackTrace();
				} catch (InterruptedException e) {
					e.printStackTrace();
				}
			}
		}.start();
		
		new Thread() {
			@Override
			public void run() {
				Zqueue queue = new Zqueue();
				try {
					Thread.sleep(3000);
				} catch (InterruptedException e) {
					e.printStackTrace();
				}
				queue.offer("hello", "ddd");
			}
		}.start();
	}

}
